package com.etlmaster.executor.steps.sql;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.HiveJdbcUtil;
import com.etlmaster.executor.utils.Regex;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HiveSqlProcessor extends SqlProcessor {

	public void runSql(String sql, JdbcTemplate jdbcTemplate) {
		jdbcTemplate.execute(sql);
		meta.log("INFO", "Sql执行成功，数据源为:{},语句：{}", meta.getDsName(), sql);
	}

	public boolean process() {
		boolean flag = true;
		// 默认执行sql语句
		String[] beforeSql = DataCollectionUtils.getProperties("beforeSql")
				.split(";");

		Connection conn = null;
		Statement stmt = null;
		try {
			// JdbcTemplate jdbcTemplate = new JdbcTemplate(meta.getDsName());

			DataSourceDefine def = DataSourceHolder.getInstance()
					.getDataSourceDefine(meta.getDsName());
			conn = HiveJdbcUtil.getConn(def.getDriverClassName(), def.getUrl(),
					def.getKrbconf(), def.getUsername(), def.getPassword());
			stmt = conn.createStatement();

			if (meta.getSetDefault().equalsIgnoreCase("true")) {
				for (String sSql : beforeSql) {
					if (!sSql.equalsIgnoreCase("null")) {
						sSql = meta.variableSubstitution(sSql);
						meta.log("INFO", sSql);
						stmt.execute(sSql);
					}
				}
			}

			if (meta.getSql() != null && meta.getSql().length() > 0) {

				String hiveSql = meta.getSql();
				if (meta.getIsOneSql().equalsIgnoreCase("false")) {
					String[] sqls = hiveSql.split("(?<!\\\\);");
					for (String sql : sqls) {
						sql = sql.trim();
						if (sql.length() > 0) {
							runSql(sql, flag, stmt);
						}
					}
				} else {
					runSql(hiveSql, flag, stmt);
				}
			} else {
				meta.log("ERROR", "配置的SQL语句为空，无法执行.或则部分变量未声明，请检查！");
				flag = false;
			}
		} catch (Exception e) {
			meta.log("ERROR", e.getMessage());
			flag = false;
		} finally {
			if (stmt != null || conn != null) {
				try {
					stmt.close();
					conn.close();
				} catch (Exception e) {
					meta.log("ERROR", "关闭连接异常：{}", e.getMessage());
				}
			}
		}

		return flag;
	}

	private void runSql(String sql, boolean flag, Statement stmt)
			throws SQLException {
		sql = sql.trim();
		if (sql.length() > 0) {
			if (!validation(sql)) {
				sql = sql.trim();
				meta.log("INFO", "begin:dbname:{} sql:{}", meta.getDsName(),
						sql);

				stmt.execute(sql);
				meta.log("INFO", "成功行数：{},数据源为:{}", 0 + "", meta.getDsName());
			} else {
				meta.log("INFO", "sql语句不合法：{},不进行执行", sql);
				flag = false;
			}
		}
	}

	private boolean validation(String sql) {
		String regex = DataCollectionUtils.getProperties("sqlRegx");
		return Regex.validation(sql, regex);

	}
}
